package com.cui.project.service.data;


import com.api.common.model.entity.GatewayAccessLog;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.cui.project.constant.RedisConstant;
import com.cui.project.mapper.GatewayAccessLogMapper;
import com.cui.project.mapper.InterfaceInfoMapper;
import com.cui.project.mapper.InterfaceTrafficMetricsMapper;
import com.cui.project.model.entity.InterfaceTrafficMetrics;
import com.cui.project.model.entity.InterfaceTrafficMetricsHourly;
import com.cui.project.model.entity.UserInterfaceTrafficMetrics;
import com.cui.project.service.InterfaceTrafficMetricsHourlyService;
import com.cui.project.service.InterfaceTrafficMetricsService;
import com.cui.project.service.UserInterfaceTrafficMetricsService;
import com.cui.project.utils.RedisBatchWriteUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 流量指标聚合服务
 * 
 * 核心功能：
 * 1. 定时聚合网关访问日志，生成流量统计指标
 * 2. 支持分钟级和小时级两种粒度的聚合
 * 3. 同步存储到MySQL（持久化）和Redis（实时查询）
 * 
 * 业务价值：
 * - 实时监控接口流量和性能
 * - 为流量预测提供历史数据基础
 * - 支持快速的流量分析和可视化
 * 
 * 聚合指标：
 * - 请求量：总请求数、成功数、失败数
 * - 性能：平均延迟、最大延迟、P95/P99延迟（仅小时级）
 * - 流量：总请求体大小
 * - 用户：独立用户数、独立IP数
 * 
 * 数据流向：
 * 网关访问日志 -> 聚合计算 -> MySQL(持久化) + Redis(缓存) -> 预测服务/资源调度
 * 
 * 技术特点：
 * - 批量写入：减少数据库压力
 * - 双写策略：MySQL保证持久化，Redis保证查询性能
 * - 事务控制：确保数据一致性
 * 
 * @author cui
 * @date 2025-11-02
 */
@Service
@Slf4j
public class TrafficMetricsAggregationService {

    /**
     * 网关访问日志数据访问对象
     * 用于查询原始访问日志
     */
    @Resource
    private GatewayAccessLogMapper gatewayAccessLogMapper;

    /**
     * 流量指标服务（分钟级）
     * 用于保存分钟级聚合数据
     */
    @Resource
    private InterfaceTrafficMetricsService trafficMetricsService;

    /**
     * 流量指标服务（小时级）
     * 用于保存小时级聚合数据
     */
    @Resource
    private InterfaceTrafficMetricsHourlyService trafficMetricsHourlyService;

    /**
     * 流量指标服务（分钟级）
     * 用于保存分钟级聚合数据
     */
    @Resource
    private UserInterfaceTrafficMetricsService userInterfaceTrafficMetricsService;

    /**
     * Redis批量写入工具
     * 优化批量写入Redis的性能
     */
    @Resource
    private RedisBatchWriteUtil redisBatchWriteUtil;

    @Resource(name = "ioExecutor")
    private Executor executor;

    @Resource
    private ApplicationContext applicationContext;

    /**
     * 最大批量处理大小：1000条
     * 避免单次事务过大，导致数据库锁等待
     */
    private static final int MAX_BATCH_SIZE=1000;


    /**
     * 聚合分钟级流量数据（定时任务）
     * 
     * 执行频率：每分钟执行一次
     * 执行时间：每分钟的第0秒（如10:00:00, 10:01:00, 10:02:00...）
     * 
     * 聚合粒度：1分钟
     * 聚合时间窗口：上一分钟的数据
     * 
     * 业务流程：
     * 1. 查询上一分钟的网关访问日志
     * 2. 按接口ID分组统计 /  用户_接口Id分组统计
     * 3. 计算每个接口的各项指标（请求数、成功率、延迟等）
     * 4. 批量写入MySQL数据库
     * 5. 批量写入Redis缓存（保留1小时）
     * 
     * 性能优化：
     * - 使用事务保证数据一致性
     * - 分批写入（每1000条），避免长事务
     * - MySQL和Redis双写，读写分离
     * 
     * 应用场景：
     * - 实时流量监控大屏
     * - 短期流量趋势分析
     * - 异常流量告警
     */
    @Scheduled(cron = "0 * * * * ?")
    public void aggregateMinuteMetrics(){
        LocalDateTime now = LocalDateTime.now();
        // 计算时间窗口：上一分钟的开始和结束时间
        // truncatedTo(MINUTES)：截断到分钟，去掉秒和毫秒
        // 示例：当前时间10:05:30 -> startTime=10:04:00, endTime=10:05:00
        LocalDateTime startTime = now.minus(1, ChronoUnit.MINUTES).truncatedTo(ChronoUnit.MINUTES);
        LocalDateTime endTime = now.truncatedTo(ChronoUnit.MINUTES);
        
        log.info("开始聚合流量数据，时间范围：{}-{}",startTime,endTime);
        try {
            // 步骤1：查询上一分钟的网关访问日志
            // 时间范围：[startTime, endTime)，左闭右开区间
            LambdaQueryWrapper<GatewayAccessLog> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.between(GatewayAccessLog::getCreatedAt,
                    Date.from(startTime.atZone(ZoneId.systemDefault()).toInstant()),
                    Date.from(endTime.atZone(ZoneId.systemDefault()).toInstant())
                    );
            List<GatewayAccessLog> logs = gatewayAccessLogMapper.selectList(queryWrapper);

            // 如果该时间段没有访问日志，直接返回
            if(logs.isEmpty()){
                log.info("该时间段内无访问日志");
                return;
            }


            // 步骤2：按接口ID分组
            // 将同一个接口的所有访问日志放在一起，便于聚合计算
            // 示例：{1001: [log1, log2, ...], 1002: [log3, log4, ...]}
            CompletableFuture<Void> f1 = CompletableFuture.runAsync(() -> {
                try {
                    Map<Long, List<GatewayAccessLog>> groupLogs = logs.stream()
                            .collect(Collectors.groupingBy(GatewayAccessLog::getInterfaceInfoId));
                    aggregationMetrics(startTime, groupLogs);
                } catch (Exception e) {
                    log.error("接口流量聚合任务异常", e);
                }
            }, executor);

            //按照userId_interfaceId分组
            // 将同一个用户使用的某个接口的所有访问日志放在一起，便于聚合计算
            // 示例：{1001_1: [log1, log2, ...], 1002_2: [log3, log4, ...]}
            CompletableFuture<Void> f2 = CompletableFuture.runAsync(() -> {
                try {
                    Map<String, List<GatewayAccessLog>> userInterfaceLogs = logs.stream()
                            .filter(l -> l.getUserId() != null)
                            .collect(Collectors.groupingBy(l -> l.getUserId() + "_" + l.getInterfaceInfoId()));
                    aggregationUserInterfaceMetrics(startTime, userInterfaceLogs);
                } catch (Exception e) {
                    log.error("用户接口流量聚合任务异常", e);
                }
            }, executor);
            //当前你启动了两个异步任务，但主线程不会等待它们完成。
            //如果定时任务频率很高（例如每分钟一次），而上一次异步任务还未执行完，就可能发生并发重叠；
            //数据聚合时间窗口如果重叠，会造成重复统计。
            CompletableFuture.allOf(f1,f2).join();
            log.info("本轮流量聚合任务全部完成");
        }catch (Exception  e){
            log.error("聚合流量数据失败", e);
            // 抛出运行时异常，触发事务回滚
            throw new RuntimeException("聚合流量数据失败", e);
        }
    }

    /**
     * 聚合用户接口的调用信息
     * @param startTime
     * @param userInterfaceLogs
     */
    public void aggregationUserInterfaceMetrics(LocalDateTime startTime, Map<String, List<GatewayAccessLog>> userInterfaceLogs) {
        log.info("开始执行用户接口流量聚合");
        // 步骤3：遍历每个接口，计算聚合指标
        List<UserInterfaceTrafficMetrics> metricsList=new ArrayList<>();
        // 用于批量写入Redis的Map，Key为Redis的key，Value为指标对象
        Map<String,Object> batchCacheMap=new LinkedHashMap<>();


        TrafficMetricsAggregationService proxy = applicationContext.getBean(TrafficMetricsAggregationService.class);
        for (Map.Entry<String, List<GatewayAccessLog>> entry : userInterfaceLogs.entrySet()) {
            String[] user_info_id = entry.getKey().split("_");
            Long userId=Long.parseLong(user_info_id[0]);
            Long interfaceId=Long.parseLong(user_info_id[1]);

            List<GatewayAccessLog> accessLogs = entry.getValue();

            // 创建流量指标对象
            UserInterfaceTrafficMetrics metrics = new UserInterfaceTrafficMetrics();
            metrics.setUserId(userId);
            metrics.setInterfaceInfoId(interfaceId);
            metrics.setTimeBucket( Date.from(startTime.atZone(ZoneId.systemDefault()).toInstant()));
            metrics.setRequestCount(accessLogs.size()); // 总请求数


            // 初始化统计变量
            int successCount=0; // 成功请求数
            long totalLatencyMs=0; // 总延迟时间（毫秒）
            int maxLatencyMs=accessLogs.get(0).getLatencyMs()!=null?accessLogs.get(0).getLatencyMs():0; // 最大延迟
            long totalBodySize=0; // 总请求体大小（字节）

            // 遍历所有访问日志，累加各项指标
            for (int i = 0; i < accessLogs.size(); i++) {
                GatewayAccessLog accessLog = accessLogs.get(i);

                // 统计成功次数（success=1表示成功）
                if(accessLog.getSuccess()==1){
                    successCount++;
                }

                // 累加延迟时间，并更新最大延迟
                if(accessLog.getLatencyMs()!=null){
                    Integer ms = accessLog.getLatencyMs();
                    totalLatencyMs+=ms;
                    if(ms>maxLatencyMs){
                        maxLatencyMs=ms;
                    }
                }

                // 累加请求体大小
                if(accessLog.getBodySize()!=null){
                    totalBodySize+=accessLog.getBodySize();
                }

            }

            // 设置成功和失败次数
            metrics.setSuccessCount(successCount);
            metrics.setFailCount(accessLogs.size()-successCount); // 失败数 = 总数 - 成功数

            // 计算平均延迟（总延迟 / 请求数）
            metrics.setAvgLatencyMs((int)(totalLatencyMs/accessLogs.size()));
            metrics.setMaxLatencyMs(maxLatencyMs);

            // 设置总请求体大小（用于流量统计和成本分析）
            metrics.setTotalBodySize(totalBodySize);

            // 统计独立用户数（去重后的用户ID数量）
            // 用于分析用户活跃度
            long uniqueUsers = accessLogs.stream()
                    .filter(l -> l.getUserId() != null)
                    .map(GatewayAccessLog::getUserId)
                    .distinct()
                    .count();
            metrics.setUniqueUsers((int)uniqueUsers);

            // 统计独立IP数（去重后的IP地址数量）
            // 用于分析访问来源、检测异常流量
            long uniqueIps = accessLogs.stream()
                    .filter(l -> l.getIp() != null)
                    .map(GatewayAccessLog::getIp)
                    .distinct()
                    .count();
            metrics.setUniqueIps((int) uniqueIps);

            // 将指标对象添加到列表，准备批量写入数据库
            metricsList.add(metrics);

            // 同时准备Redis缓存数据
            // Key格式：api:traffic:minutes:1001:2025-11-02T10:30
            batchCacheMap.put(String.format(RedisConstant.TRAFFIC_USER_INTERFACE_METRICS_MINUTES_CACHED_KEY,
                            metrics.getUserId(),
                            metrics.getInterfaceInfoId(),
                            startTime.toString()),
                    metrics);
        }

        // 步骤4：批量插入到MySQL数据库
        // 分批处理，每次最多1000条，避免以下问题：
        // - 单次事务过大，导致数据库锁等待
        // - 内存占用过高
        // - 事务超时
        for (int i = 0; i < metricsList.size(); i+=MAX_BATCH_SIZE) {
           proxy.saveUserInterfaceBatchWithTransaction(metricsList.subList(i,Math.min(i+MAX_BATCH_SIZE,metricsList.size())));
        }

        // 步骤5：批量写入Redis缓存
        // 用途：
        // - 实时监控大屏快速查询
        // - 减轻MySQL查询压力
        // 过期时间：1小时（分钟级数据仅需短期缓存）
        redisBatchWriteUtil.batchSetWithExpire(batchCacheMap,1, TimeUnit.HOURS);

        log.info("用户接口流量数据聚合完成，共聚合 {} 条用户接口记录", metricsList.size());
    }

    /**
     * 聚合接口在某个时间时总调用信息
     * @param startTime
     * @param groupLogs
     */
    public void aggregationMetrics(LocalDateTime startTime, Map<Long, List<GatewayAccessLog>> groupLogs) {
        log.info("开始执行接口流量聚合");
        // 步骤3：遍历每个接口，计算聚合指标
        List<InterfaceTrafficMetrics> metricsList=new ArrayList<>();
        // 用于批量写入Redis的Map，Key为Redis的key，Value为指标对象
        Map<String,Object> batchCacheMap=new LinkedHashMap<>();

        TrafficMetricsAggregationService proxy = applicationContext.getBean(TrafficMetricsAggregationService.class);
        for (Map.Entry<Long, List<GatewayAccessLog>> entry : groupLogs.entrySet()) {
            Long interfaceId = entry.getKey();
            List<GatewayAccessLog> accessLogs = entry.getValue();

            // 创建流量指标对象
            InterfaceTrafficMetrics metrics = new InterfaceTrafficMetrics();
            metrics.setInterfaceInfoId(interfaceId);
            metrics.setTimeBucket( Date.from(startTime.atZone(ZoneId.systemDefault()).toInstant()));
            metrics.setRequestCount(accessLogs.size()); // 总请求数


            // 初始化统计变量
            int successCount=0; // 成功请求数
            long totalLatencyMs=0; // 总延迟时间（毫秒）
            int maxLatencyMs=accessLogs.get(0).getLatencyMs()!=null?accessLogs.get(0).getLatencyMs():0; // 最大延迟
            long totalBodySize=0; // 总请求体大小（字节）

            // 遍历所有访问日志，累加各项指标
            for (int i = 0; i < accessLogs.size(); i++) {
                GatewayAccessLog accessLog = accessLogs.get(i);

                // 统计成功次数（success=1表示成功）
                if(accessLog.getSuccess()==1){
                    successCount++;
                }

                // 累加延迟时间，并更新最大延迟
                if(accessLog.getLatencyMs()!=null){
                    Integer ms = accessLog.getLatencyMs();
                    totalLatencyMs+=ms;
                    if(ms>maxLatencyMs){
                        maxLatencyMs=ms;
                    }
                }

                // 累加请求体大小
                if(accessLog.getBodySize()!=null){
                    totalBodySize+=accessLog.getBodySize();
                }

            }

            // 设置成功和失败次数
            metrics.setSuccessCount(successCount);
            metrics.setFailCount(accessLogs.size()-successCount); // 失败数 = 总数 - 成功数

            // 计算平均延迟（总延迟 / 请求数）
            metrics.setAvgLatencyMs((int)(totalLatencyMs/accessLogs.size()));
            metrics.setMaxLatencyMs(maxLatencyMs);

            // 设置总请求体大小（用于流量统计和成本分析）
            metrics.setTotalBodySize(totalBodySize);

            // 统计独立用户数（去重后的用户ID数量）
            // 用于分析用户活跃度
            long uniqueUsers = accessLogs.stream()
                    .filter(l -> l.getUserId() != null)
                    .map(GatewayAccessLog::getUserId)
                    .distinct()
                    .count();
            metrics.setUniqueUsers((int)uniqueUsers);

            // 统计独立IP数（去重后的IP地址数量）
            // 用于分析访问来源、检测异常流量
            long uniqueIps = accessLogs.stream()
                    .filter(l -> l.getIp() != null)
                    .map(GatewayAccessLog::getIp)
                    .distinct()
                    .count();
            metrics.setUniqueIps((int) uniqueIps);

            // 将指标对象添加到列表，准备批量写入数据库
            metricsList.add(metrics);

            // 同时准备Redis缓存数据
            // Key格式：api:traffic:minutes:1001:2025-11-02T10:30
            batchCacheMap.put(String.format(RedisConstant.TRAFFIC_METRICS_MINUTES_CACHED_KEY,
                        metrics.getInterfaceInfoId(),
                        startTime.toString()),
                    metrics);
        }

        // 步骤4：批量插入到MySQL数据库
        // 分批处理，每次最多1000条，避免以下问题：
        // - 单次事务过大，导致数据库锁等待
        // - 内存占用过高
        // - 事务超时
        for (int i = 0; i < metricsList.size(); i+=MAX_BATCH_SIZE) {
            proxy.saveInterfaceBatchWithTransaction(metricsList.subList(i,Math.min(i+MAX_BATCH_SIZE,metricsList.size())));
        }

        // 步骤5：批量写入Redis缓存
        // 用途：
        // - 实时监控大屏快速查询
        // - 减轻MySQL查询压力
        // 过期时间：1小时（分钟级数据仅需短期缓存）
        redisBatchWriteUtil.batchSetWithExpire(batchCacheMap,1, TimeUnit.HOURS);


        log.info("流量数据聚合完成，共聚合 {} 个接口的数据", metricsList.size());
    }
    /**
     * ✅ 批量保存接口流量数据（带事务）
     * 必须是 public 方法，才能被Spring代理
     */
    @Transactional(rollbackFor = Exception.class)
    public void saveInterfaceBatchWithTransaction(List<InterfaceTrafficMetrics> batch) {
        trafficMetricsService.saveBatch(batch);
        log.debug("批次保存成功，数量：{}", batch.size());
    }

    /**
     * ✅ 批量保存接口流量数据（带事务）
     * 必须是 public 方法，才能被Spring代理
     */
    @Transactional(rollbackFor = Exception.class)
    public void saveUserInterfaceBatchWithTransaction(List<UserInterfaceTrafficMetrics> batch) {
        userInterfaceTrafficMetricsService.saveBatch(batch);
        log.debug("批次保存成功，数量：{}", batch.size());
    }

    /**
     * 聚合小时级流量数据（定时任务）
     *
     * 执行频率：每小时执行一次
     * 执行时间：每小时的第5分钟（如10:05:00, 11:05:00, 12:05:00...）
     * 执行时间选择原因：
     * - 避开整点时刻，降低系统负载
     * - 留出5分钟的缓冲时间，确保上一小时的数据已完全写入
     *
     * 聚合粒度：1小时
     * 聚合时间窗口：上一小时的数据
     *
     * 与分钟级聚合的区别：
     * - 时间粒度更粗，数据量更小，适合长期存储
     * - 增加了P95、P99延迟指标（需要排序计算）
     * - 更适合用于流量预测和趋势分析
     *
     * 业务流程：
     * 1. 查询上一小时的网关访问日志
     * 2. 按延迟升序排列（用于计算百分位数）
     * 3. 按接口ID分组统计
     * 4. 计算各项指标，包括P95、P99延迟
     * 5. 批量写入MySQL和Redis
     *
     * 应用场景：
     * - 流量预测的数据来源
     * - 长期趋势分析
     * - 性能基线建立
     */
    @Scheduled(cron = "0 5 * * * ?")
    @Transactional(rollbackFor = Exception.class)
    public void aggregateHourlyMetrics(){
        LocalDateTime now = LocalDateTime.now();
        // 计算时间窗口：上一小时的开始和结束时间
        // 示例：当前时间10:05:00 -> startTime=09:00:00, endTime=10:00:00
        LocalDateTime startTime = now.minus(1, ChronoUnit.HOURS).truncatedTo(ChronoUnit.HOURS);
        LocalDateTime endTime = now.truncatedTo(ChronoUnit.HOURS);

        log.info("开始聚合小时级流量数据，时间范围：{}-{}",startTime,endTime);
        try {
            // 步骤1：查询上一小时的网关访问日志
            // 按延迟升序排列，用于后续计算P95、P99延迟
            LambdaQueryWrapper<GatewayAccessLog> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.between(GatewayAccessLog::getCreatedAt,
                    Date.from(startTime.atZone(ZoneId.systemDefault()).toInstant()),
                    Date.from(endTime.atZone(ZoneId.systemDefault()).toInstant())
            );
            //数据库层面对所有日志排序，如果数据量大会很慢
//                    .orderByAsc(GatewayAccessLog::getLatencyMs); // 按延迟升序排列
            List<GatewayAccessLog> logs = gatewayAccessLogMapper.selectList(queryWrapper);

            // 如果该时间段没有访问日志，直接返回
            if(logs.isEmpty()){
                log.info("该时间段内无访问日志");
                return;
            }

            // 步骤2：按接口ID分组
            Map<Long, List<GatewayAccessLog>> groupLogs = logs.stream().collect(Collectors.groupingBy(GatewayAccessLog::getInterfaceInfoId));

            // 步骤3：遍历每个接口，计算聚合指标
            List<InterfaceTrafficMetricsHourly> metricsList=new ArrayList<>();
            Map<String,Object> batchCacheMap=new LinkedHashMap<>();

            for (Map.Entry<Long, List<GatewayAccessLog>> entry : groupLogs.entrySet()) {
                Long interfaceId = entry.getKey();
                List<GatewayAccessLog> accessLogs = entry.getValue();

                // 不在数据库层排序，而是在分组后对每个接口的日志排序
                // 对当前接口的日志按延迟排序
                accessLogs.sort(Comparator.comparing(
                        l->l.getLatencyMs()!=null?l.getLatencyMs():0
                ));

                // 创建小时级流量指标对象
                InterfaceTrafficMetricsHourly metricsHourly=new InterfaceTrafficMetricsHourly();
                metricsHourly.setInterfaceInfoId(interfaceId);
                metricsHourly.setTimeBucket( Date.from(startTime.atZone(ZoneId.systemDefault()).toInstant()));
                int size = accessLogs.size();
                metricsHourly.setRequestCount(size); // 总请求数

                // 初始化统计变量
                int successCount=0;
                long totalLatencyMs=0;
                long totalBodySize=0;
                
                // 遍历所有访问日志，累加各项指标
                for (int i = 0; i < size; i++) {
                    GatewayAccessLog accessLog = accessLogs.get(i);
                    
                    // 统计成功次数
                    if(accessLog.getSuccess()==1){
                        successCount++;
                    }
                    
                    // 累加延迟时间
                    if(accessLog.getLatencyMs()!=null){
                        totalLatencyMs+=accessLog.getLatencyMs();
                    }
                    
                    // 累加请求体大小
                    if(accessLog.getBodySize()!=null){
                        totalBodySize+=accessLog.getBodySize();
                    }
                }
                
                // 设置成功和失败次数
                metricsHourly.setSuccessCount(successCount);
                metricsHourly.setFailCount(accessLogs.size()-successCount);

                // 计算平均延迟
                metricsHourly.setAvgLatencyMs((int)(totalLatencyMs/accessLogs.size()));

                // 设置总请求体大小
                metricsHourly.setTotalBodySize(totalBodySize);

                // 统计独立用户数
                long uniqueUsers = accessLogs.stream()
                        .filter(l -> l.getUserId() != null)
                        .map(GatewayAccessLog::getUserId)
                        .distinct()
                        .count();
                metricsHourly.setUniqueUsers((int)uniqueUsers);

                // 统计独立IP数
                long uniqueIps = accessLogs.stream()
                        .filter(l -> l.getIp() != null)
                        .map(GatewayAccessLog::getIp)
                        .distinct()
                        .count();
                metricsHourly.setUniqueIps((int) uniqueIps);


                // 计算P95响应时间（95%的请求延迟低于此值）
                // P95是性能监控的重要指标，比平均值更能反映用户体验
                // 公式：取排序后第95%位置的值
                // 注意：这里的计算有误，应该是 accessLogs.get((int)Math.ceil(size*0.95)-1).getLatencyMs()
                int p95Index = (int)Math.ceil(size * 0.95) - 1;
                if (p95Index >= 0 && p95Index < accessLogs.size()) {
                    Integer p95Latency = accessLogs.get(p95Index).getLatencyMs();
                    metricsHourly.setP95LatencyMs(p95Latency != null ? p95Latency : 0);
                }
                
                // 计算P99响应时间（99%的请求延迟低于此值）
                // P99用于发现长尾延迟问题
                int p99Index = (int)Math.ceil(size * 0.99) - 1;
                if (p99Index >= 0 && p99Index < accessLogs.size()) {
                    Integer p99Latency = accessLogs.get(p99Index).getLatencyMs();
                    metricsHourly.setP99LatencyMs(p99Latency != null ? p99Latency : 0);
                }

                // 将指标对象添加到列表
                metricsList.add(metricsHourly);
                
                // 准备Redis缓存数据
                batchCacheMap.put(String.format(RedisConstant.TRAFFIC_METRICS_HOURLY_CACHED_KEY,
                                metricsHourly.getInterfaceInfoId(),
                                startTime.toString()),
                        metricsHourly);
            }

            // 步骤4：批量插入到MySQL数据库
            // 分批处理，避免长事务
            for (int i = 0; i < metricsList.size(); i+=MAX_BATCH_SIZE) {
                trafficMetricsHourlyService.saveBatch(metricsList.subList(i,Math.min(i+MAX_BATCH_SIZE,metricsList.size())));
            }

            // 步骤5：批量写入Redis缓存
            // 过期时间：1小时
            // 注意：小时级数据实际上可以缓存更长时间（如24小时），因为查询频率较低
            redisBatchWriteUtil.batchSetWithExpire(batchCacheMap,1, TimeUnit.HOURS);

            log.info("小时级流量数据聚合完成，共聚合 {} 个接口的数据", metricsList.size());
        }catch (Exception  e){
            log.error("小时级聚合流量数据失败", e);
            // 抛出运行时异常，触发事务回滚
            throw new RuntimeException("小时级聚合流量数据失败", e);
        }
    }
    
    /**
     * 测试方法：验证时间计算逻辑
     * 仅用于开发调试，生产环境应删除
     */
    public static void main(String[] args) {
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime startTime = now.minus(1, ChronoUnit.HOURS).truncatedTo(ChronoUnit.HOURS);
        LocalDateTime endTime = now.truncatedTo(ChronoUnit.HOURS);
        System.out.println(startTime);
        System.out.println(endTime);
        System.out.println( Date.from(startTime.atZone(ZoneId.systemDefault()).toInstant()));
    }
}
